package com.fxkj.cloud.drms.ui.emqx;

import com.fxkj.cloud.drms.ui.emqx.model.EmqxClientInfo;
import com.fxkj.cloud.drms.ui.emqx.model.EmqxLoginResult;
import com.fxkj.cloud.drms.ui.emqx.model.EmqxResponse;
import com.fxkj.cloud.drms.ui.emqx.model.EmqxUser;
import com.fxkj.cloud.drms.infra.config.MqttProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.support.WebClientAdapter;
import org.springframework.web.service.invoker.HttpServiceProxyFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * <p>Description: </p>
 *
 * @author Wei, Wu
 * @create 2024/9/26 13:23
 */
@Component
@Slf4j
public class EmqxServer implements ApplicationListener<ApplicationReadyEvent> {
    private final EmqxClient emqxClient;
    private EmqxAuthorizedClient emqxAuthorizedClient;
    private final String host;
    private final Integer port;
    private final String username;
    private final String password;
    private String token;

    public EmqxServer(EmqxClient emqxClient,
                      MqttProperties mqttProperties) {
        this.emqxClient = emqxClient;
        this.host = mqttProperties.getHost();
        this.port = mqttProperties.getWeb().getPort();
        this.username = mqttProperties.getWeb().getUsername();
        this.password = mqttProperties.getWeb().getPassword();
        this.initAuthorizationClient();
    }

    public boolean initAuthorizationClient() {
        try {
            EmqxLoginResult loginResult = emqxClient.login(new EmqxUser(username, password));
            if (loginResult == null || loginResult.getToken() == null) {
                return false;
            }
            token = loginResult.getToken();
            this.emqxAuthorizedClient = this.emqxAuthorizedClient(host, port, token);
            log.info("EMQX客户端认证成功,token:{}", token);
            return true;
        }catch (Exception e) {
            log.error("EMQX客户端认证失败", e);
            return false;
        }
    }

    private EmqxAuthorizedClient emqxAuthorizedClient(String host, Integer port, String token) {
        WebClient webClient = WebClient.builder()
                .baseUrl("http://" + host + ":" + port)
                .defaultHeader("Authorization", "Bearer " + token)
                .build();
        HttpServiceProxyFactory httpServiceProxyFactory =
                HttpServiceProxyFactory.builderFor(WebClientAdapter.create(webClient))
                        .build();
        return httpServiceProxyFactory.createClient(EmqxAuthorizedClient.class);
    }

    public EmqxAuthorizedClient getEmqxAuthorizedClient() {
        return this.emqxAuthorizedClient;
    }

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        //过滤clientId条件,仅查找设备客户端
        String likeClientId = "fxkj-device/";
        //设置返回条数
        Integer limit = 10000;
        //设置返回字段
        List<String> fields = new ArrayList<>();
        fields.add("clientid");
        fields.add("connected");
        String fieldsStr = fields.stream().collect(Collectors.joining(","));
        //设置页数
        Integer page = 1;

        List<EmqxClientInfo> onlineClients = new ArrayList<>(10240);
        //获取在线客户端列表
        EmqxResponse<List<EmqxClientInfo>> onlineClientPage = emqxAuthorizedClient.getOnlineClient(likeClientId, limit, page, fieldsStr);
        onlineClients.addAll(onlineClientPage.getData().stream().filter(EmqxClientInfo::getConnected).collect(Collectors.toSet()));
        if (onlineClientPage.getMeta().getHasnext()) {
            page++;
            while (true) {
                //先限制条数为10w条,避免死循环
                if (page > 10) {
                    break;
                }
                onlineClientPage = emqxAuthorizedClient.getOnlineClient(likeClientId, limit, page, fieldsStr);
                onlineClients.addAll(onlineClientPage.getData().stream().filter(EmqxClientInfo::getConnected).collect(Collectors.toSet()));
            }
        }
        log.info("EMQX在线客户端数:{}", onlineClients.size());
        //todo 推送设备在线列表

    }
}
